package com.csw.flink.kafka

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.io.Source


/**
  * kafka中数据是以kv格式存储的，但一般用不上k
  */
object Demo01Produce {
  def main(args: Array[String]): Unit = {

    /**
      * 创建生产者
      */

    val properties: Properties = new Properties

    //kafka  broker 地址列表
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    //key和value序列化的类
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    val producer: KafkaProducer[String, String] = new KafkaProducer[String, String](properties)

    /**
      * 生产数据
      */

    //读取学生表
    val students: List[String] = Source.fromFile("flink/data/students.txt").getLines().toList

    for (student <- students) {

      /**
        * topic可以不存在，在生产数据时会自动创建，自动创建一个分区是为1，副本数为1的topic
        *
        */
      //将同一个班级的数据存储到同一个分区
      //好处： 后续按照班级分组效率高
      //缺点： 可能会导致数据倾斜
      //kafka-topics.sh --create --zookeeper master:2181,node1:2181,node2:2181 --replication-factor 3 --partitions 3 --topic student1

      //获取到班级
      val clazz: String = student.split(",")(4)

      //hash   计算班级属于哪一个分区
      val partition: Int = Math.abs(clazz.hashCode) % 3
      //参数分别为:topic,分区，时间戳，key，value
      val record = new ProducerRecord[String, String]("student1", partition,null,student)

      //发送数据
      producer.send(record)

      //刷新到kafka集群
      producer.flush()


    }

    //关闭连接
    producer.close()
  }
}
